package com.demo.flink.monitor;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

/**
 * @author : pengjie
 * @PackageName : com.demo.flink.monitor
 * @Description : TODO
 * @email : 627799251@qq.com
 * @Date : 2019/1/31 18:25
 */
public class MonitorMain {

    // 参考链接 https://cloud.tencent.com/developer/article/1368819

    private static final double TEMPERATURE_THRESHOLD = 60.0;
    private static final double TEMPERATURE_THRESHOLD2 = 80.0;

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<MonitoringEvent> inputEventStream = env.addSource(new SourceFunction<MonitoringEvent>() {
            @Override
            public void run(SourceContext<MonitoringEvent> sourceContext) throws Exception {
                while (true){
                    TemperatureEvent event = new TemperatureEvent();
                    event.setRackID((int)(Math.random()*10));
                    event.setTemperature((Math.random()*100));

                    TemperatureEvent event2 = new TemperatureEvent();
                    event2.setRackID((int)(Math.random()*10));
                    event2.setTemperature((Math.random()*100));

                    sourceContext.collect(event);
                    sourceContext.collect(event2);
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });

        Pattern<MonitoringEvent, ?> warningPattern = Pattern.<MonitoringEvent>begin("First Event")
                .subtype(TemperatureEvent.class)
                .where(new SimpleCondition<TemperatureEvent>() {
                    @Override
                    public boolean filter(TemperatureEvent temperatureEvent) throws Exception {
                        return temperatureEvent.getTemperature() > TEMPERATURE_THRESHOLD;
                    }
                }).next("Second Event")
                .subtype(TemperatureEvent.class)
                .where(new SimpleCondition<TemperatureEvent>() {
                    @Override
                    public boolean filter(TemperatureEvent temperatureEvent) throws Exception {
                        return temperatureEvent.getTemperature() > TEMPERATURE_THRESHOLD2;
                    }
                })
                .within(Time.seconds(10));

        //KeyedStream<MonitoringEvent, Tuple> keyedStream  = inputEventStream.keyBy("rackId");

        PatternStream<MonitoringEvent> tempPatternStream = CEP.pattern(inputEventStream.keyBy("rackID"), warningPattern);

        DataStream<TemperatureWarning> warnings = tempPatternStream.select(new PatternSelectFunction<MonitoringEvent, TemperatureWarning>() {
            @Override
            public TemperatureWarning select(Map<String, List<MonitoringEvent>> pattern) throws Exception {
                //System.out.println("TemperatureWarning : pattern First Event list = " + pattern.get("First Event").size());
                //System.out.println("TemperatureWarning : pattern Second Event list = " + pattern.get("Second Event").size());

                TemperatureEvent first = (TemperatureEvent) pattern.get("First Event").get(0);
                TemperatureEvent second = (TemperatureEvent) pattern.get("Second Event").get(0);

                System.out.println("rackID = " + first.getRackID());
                System.out.println("first = " + first.toString());
                System.out.println("second = " + second.toString());

                return new TemperatureWarning(first.getRackID(), (first.getTemperature() + second.getTemperature()) / 2);
            }
        });

        warnings.addSink(new SinkFunction<TemperatureWarning>() {
            @Override
            public void invoke(TemperatureWarning value, Context context) throws Exception {
                //System.out.println("发生警告， " + value.toString());
            }
        });

        Pattern<TemperatureWarning, ?> alertPattern = Pattern.<TemperatureWarning>begin("First Event")
                .next("Second Event")
                .within(Time.seconds(3000));

        PatternStream<TemperatureWarning> alertPatternStream = CEP.pattern(warnings.keyBy("rackID"), alertPattern);


        DataStream<TemperatureAlert> alerts = alertPatternStream.flatSelect(new PatternFlatSelectFunction<TemperatureWarning, TemperatureAlert>() {
            @Override
            public void flatSelect(Map<String, List<TemperatureWarning>> pattern, Collector<TemperatureAlert> collector) throws Exception {

                //System.out.println("TemperatureAlert : pattern First Event list = " + pattern.get("First Event").size());
                //System.out.println("TemperatureAlert : pattern Second Event list = " + pattern.get("Second Event").size());

                TemperatureWarning first = pattern.get("First Event").get(0);
                TemperatureWarning second = pattern.get("Second Event").get(0);

                if (first.getAverageTemperature() < second.getAverageTemperature()) {
                    collector.collect(new TemperatureAlert(first.getRackID()));
                }
            }
        });

        alerts.addSink(new SinkFunction<TemperatureAlert>() {
            private Logger logger = LoggerFactory.getLogger(this.getClass());

            @Override
            public void invoke(TemperatureAlert value, Context context) throws Exception {
                logger.info("机器出现严重报警 ： " + value.toString());
            }
        });

        env.execute("Monitor Demo");
    }
}
